package com.azh.server.datam;

import com.azh.server.common.AbstractQuoteCache;
import com.azh.server.datam.dataItem.DataItem;
import com.azh.server.datam.dataItem.DataItemImpl;
import com.azh.server.datam.logger.Logger;
import com.azh.server.datam.page.Page;
import com.azh.server.datam.page.PageMeta;
import com.azh.server.datam.page.PageNormal;
import com.azh.server.datam.pagecache.PageCache;
import com.azh.server.datam.indexs.PageIndex;
import com.azh.server.datam.indexs.PageInfo;
import com.azh.server.tx.TxManager;
import com.azh.server.util.ErrorUtil;
import com.azh.server.util.ExceptionUtils;
import com.azh.server.util.Types;

/**
 * @author zhaozhenhang <zhaozhenhang@kuaishou.com>
 * Created on 2022-04-26
 */
public class DataManagerImpl extends AbstractQuoteCache<DataItem> implements DataManager{

    private TxManager tx;
    private PageCache pageCache;
    private Logger logger;
    private PageIndex pageIndex;
    private Page pageOne;

    public DataManagerImpl(PageCache pageCache, Logger logger, TxManager tx) {
        super(0);
        this.pageCache = pageCache;
        this.logger = logger;
        this.tx = tx;
        this.pageIndex = new PageIndex();
    }

    /**
     * 根据uid从缓存中
     * @param uid
     * @return
     * @throws Exception
     */
    @Override
    public DataItem read(long uid) throws Exception {
        DataItemImpl dii = (DataItemImpl) super.get(uid);
        if (!dii.isValid()) {
            dii.release();
            return null;
        }
        return dii;
    }

    /**
     * 在 pageIndex 中获取一个足以存储插入内容的页面的页号，获取后
     * 先插入日志，然后通过pageNormal插入数据，最后返回插入位置的偏移，然后回写pageIndex
     * @param txId
     * @param data
     * @return
     * @throws Exception
     */
    @Override
    public long insert(long txId, byte[] data) throws Exception {
        byte[] raw = DataItem.wrapDataItemRaw(data);
        if (raw.length > PageNormal.max_free_space) {
            ErrorUtil.error(new RuntimeException("Data too large!"));
        }
        PageInfo info = null;
        for (int i = 0; i < 5; i++) { //todo 为啥是5
            info = pageIndex.select(raw.length);
            if (info != null) {
                break;
            }
            int newPageNumber = pageCache.newPage(PageNormal.initRaw());
            pageIndex.add(newPageNumber, PageNormal.max_free_space);
        }
        if (info == null) {
            ErrorUtil.error(new RuntimeException("Database is busy!"));
        }
        Page page = null;
        int freeSpace = 0;
        try {
            page = pageCache.getPage(info.getPageNumber());
            byte[] log = Recover.insertLog(txId, page, raw);
            logger.log(log);
            short offset = PageNormal.insert(page, raw);
            page.release();
            return Types.addressToUid(info.getPageNumber(), offset);
        }finally {
            if (page != null) {
                pageIndex.add(info.getPageNumber(), PageNormal.getFreeSpace(page));
            }else {
                pageIndex.add(info.getPageNumber(), freeSpace);
            }
        }
    }

    /**
     * DataManager关闭时，需要执行缓存和日志的关闭流程，设置第一页的字节校验
     */
    @Override
    public void close() {
        super.close();
        logger.close();
        PageMeta.setVcClose(pageOne);
        pageOne.release();
        pageCache.close();
    }

    /**
     * 在打开文件是校验meta的正确性，主要是为了奔溃恢复
     * @return
     */
    public boolean loadCheckPageMeta() {
        try {
            pageOne = pageCache.getPage(1);
        } catch (Exception e) {
            ExceptionUtils.getMessage(e);
        }
        return PageMeta.checkVc(pageOne);
    }

    /**
     *
     * DataManager 是 DM 层直接对外提供方法的类，同时，也实现成 DataItem 对象的缓存。DataItem 存储的 key，
     * 是由页号和页内偏移组成的一个 8 字节无符号整数，页号和偏移各占 4 字节。
     * 需要从key中解析出页号，从pageCache中获取到页面，根据偏移，解析出DataItem
     * @param key
     * @return
     * @throws Exception
     */
    @Override
    protected DataItem getForDB(long key) throws Exception {
        short offset = (short) (key & (1L << 16));
        key >>>= 32;
        int pageNumber = (int) (key & ((1L << 32) - 1));
        Page page = pageCache.getPage(pageNumber);
        return DataItem.parseDataItem(page, offset, this);
    }

    /**
     * DataItem释放时，需要将DataItem回写进入db，对dataItem以页为单位释放
     * @param obj
     */
    @Override
    protected void releaseForDB(DataItem obj) {
        obj.page().release();
    }

    /**
     * DataManager被创建时，需要获取所有页面并填充PageIndex
     */
    public void fillPageIndex() {
        int pageNumber = pageCache.getPageNumber();
        for (int i = 2; i <= pageNumber; i++) {
            Page page = null;
            try {
                page = pageCache.getPage(i);
            } catch (Exception e) {
                ExceptionUtils.getMessage(e);
            }
            pageIndex.add(page.getPageNumber(), PageNormal.getFreeSpace(page));
            // 使用完page 后及时release
            page.release();
        }
    }

    /**
     * 为txId生成update日志
     * @param txId
     * @param di
     */
    public void logDataItem(long txId, DataItem di) {
        byte[] log = Recover.updateLog(txId, di);
        logger.log(log);
    }

    public void releaseDataItem(DataItem dataItem) {
        super.release(dataItem.getUid());
    }

    /**
     * 初始化第一页，也就是元数据
     */
    public void initPageMeta() {
        int pageNumber = pageCache.newPage(PageMeta.initRaw());
        try {
            pageOne = pageCache.getPage(pageNumber);
        } catch (Exception e) {
            ExceptionUtils.getMessage(e);
        }
        pageCache.flushPage(pageOne);
    }

    public TxManager getTx() {
        return tx;
    }

    public void setTx(TxManager tx) {
        this.tx = tx;
    }

    public PageCache getPageCache() {
        return pageCache;
    }

    public void setPageCache(PageCache pageCache) {
        this.pageCache = pageCache;
    }

    public Logger getLogger() {
        return logger;
    }

    public void setLogger(Logger logger) {
        this.logger = logger;
    }

    public PageIndex getPageIndex() {
        return pageIndex;
    }

    public void setPageIndex(PageIndex pageIndex) {
        this.pageIndex = pageIndex;
    }

    public Page getPageOne() {
        return pageOne;
    }

    public void setPageOne(Page pageOne) {
        this.pageOne = pageOne;
    }
}
